package com.gmall.app

import java.time.{Instant, LocalDateTime, ZoneId}
import java.time.format.DateTimeFormatter
import java.util

import com.gmall.bean.{AlertInfo, EventLog}
import com.gmall.utils.Constants
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.DStream
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods

import scala.util.control.Breaks._
import com.gmall.utils.MyESUtil._

/**
 * <p>文件名称: 实时预警 </p>
 * <p>描述: [类型描述] </p>
 * <p>创建时间: 2020/10/21 </p>
 * @author <a href="mail to: 1034652894@qq.com" rel="nofollow">Demik Qin</a>
 * @version v1.0
 * @update [序号][日期YYYY-MM-DD] [更改人姓名][变更描述]
 */
object AlertApp extends BaseApp {
    override val appName: String = "AlertApp"
    override val master: String = "local[*]"
    override val batchTime: Int = 3
    override val groupId: String = "AlertApp"
    override val topic: String = Constants.EVENT_TOPIC

    /*
        什么样的行为需要预警:

        需求：
        同一设备，5分钟内三次及以上用不同账号登录并领取优惠劵，
        并且在登录到领劵过程中没有浏览商品。
        同时达到以上要求则产生一条预警日志。
        同一设备，每分钟只记录一次预警。

        分析:
        同一设备   按照设备id分组
        5分钟内    window(窗口长度: 5分钟  滑动步长: 6s)

        三次及以上用不同账号登录并领取优惠劵
            即：统计领取优惠券的用户的个数
        并且在登录到领劵过程中没有浏览商品
            没有浏览商品: 5分种内没有出现浏览商品的行为

        es处理:
            同一设备，每分钟只记录一次预警。
        spark-Streaming不负责, 交给es来完成
 */

    override def run(ssc: StreamingContext, sourceStream: DStream[String]): Unit = {
        sourceStream
          // 添加窗口（窗口长度: 5分钟  滑动步长: 6s）
          .window(Seconds(5 * 60), Seconds(6))
          // 解析json，并封装成二元元组(mid,eventLog)
          .map(json => {
              implicit val format: DefaultFormats.type = org.json4s.DefaultFormats
              val log: EventLog = JsonMethods.parse(json).extract[EventLog]
              val ldt = LocalDateTime.ofInstant(Instant.ofEpochMilli(log.ts), ZoneId.systemDefault())
              val yyyyMMddFormat = DateTimeFormatter.ofPattern("yyyy-MM-dd")
              val HHFormat = DateTimeFormatter.ofPattern("HH")
              log.logDate = ldt.format(yyyyMMddFormat)
              log.logHour = ldt.format(HHFormat)
              (log.mid, log)
          })
          // 按照 mid 分组
          .groupByKey()
          // 预警的业务逻辑
          .map {
              case (mid, iter) =>
                  // 1. 统计领取优惠券用户的个数量
                  val uids: util.HashSet[String] = new java.util.HashSet[String]()
                  // 2. 当前设备所有的行为
                  val events: util.ArrayList[String] = new java.util.ArrayList[String]()
                  // 3. 优惠券对应的商品id
                  val items: util.HashSet[String] = new java.util.HashSet[String]()
                  // 4. 标记是否浏览过商品. 默认没有
                  var isBrowser: Boolean = false
                  breakable {
                      iter.foreach(eventLog => {
                          events.add(eventLog.eventId)
                          eventLog.eventId match {
                              case "coupon" =>
                                  uids.add(eventLog.uid) // 把领取优惠券的用户的id存起来
                                  items.add(eventLog.itemId) // 存储优惠券对应的商品id
                              case "clickItem" =>
                                  isBrowser = true
                                  break
                              case _ =>
                          }
                      })
                  }
                  // 组合成元组  (是否预警true/false, 预警信息)
                  (!isBrowser && uids.size() > 3,
                    AlertInfo(mid, uids, items, events, System.currentTimeMillis()))
          }
          .filter(_._1) // 过滤出预警警告的信息
          .map(_._2) // 取出eventLog日志
          .foreachRDD(rdd=>{
              println("xxxxxx")
              // 把预警信息写入到ES
              rdd.saveToES("gmall_coupon_alert")
          })
    }
}
